package cn.admin.remoting.impl;

import cn.admin.remoting.*;
import cn.admin.remoting.locale.LogResources;
import cn.admin.remoting.util.*;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.pool.impl.GenericObjectPool;

import java.net.InetSocketAddress;
import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;

public class DefaultClientManager extends ClientManager {

    public static final TrScheduledExecutor timer = new TrScheduledExecutor("cn.admin.remoting" +
            ".TimerThread");
    final ReconnectManager reconnectMgr = new ReconnectManager();
    final RepositoryManager repoMgr = new RepositoryManager();

    @Override
    public Client get(String appName, String targetUrl,
                      InetSocketAddress localAddress, List<IOEventListener> listeners,
                      Map<Class<?>, RequestProcessor<?>> processors) throws RemotingException, InterruptedException {
        ConnectionUrl urlObj = ConnectionUrl.get(targetUrl);
        List<IOEventListener> copyListeners = copyListener(listeners,processors);
        int count = Integer.parseInt(urlObj.getProperty("_CONNECTIONNUM"));
        if (count == 1) {
            try {
                return this.getAsync(appName, targetUrl, localAddress, copyListeners, null).get(-1L);
            } catch (RemotingException e) {
                this.reconnectMgr.addReconnectTask(appName,targetUrl,localAddress,copyListeners);
                throw e;
            }
        } else {
            String key = this.repoMgr.getClientPoolKey(appName,targetUrl);
            try {
                GenericObjectPool<Object> clientPool = this.repoMgr.getClientPool(key,
                        copyListeners);
                synchronized (clientPool) {
                    Client client = (Client) clientPool.borrowObject();
                    clientPool.returnObject(client);
                    return client;
                }
            } catch (Exception e) {
                this.reconnectMgr.addReconnectTask(appName,targetUrl,localAddress,copyListeners);
                String msg = LogResources.getLog(LogResources.GET_CONNECT_FAIL,targetUrl);
                throw new RemotingException(msg,e);
            }
        }
    }

    @Override
    public ClientFuture getAsync(String appName, String targetUrl, InetSocketAddress localAddress
            ,List<IOEventListener> listeners, Map<Class<?>, RequestProcessor<?>> processors) {
        List<IOEventListener> copy = copyListener(listeners,processors);
        String key = ClientKeyUtil.singleClientKey(appName,targetUrl,localAddress);
        return this.repoMgr.getSingleClientAsync(key,copy);
    }

    @Override
    public void closeEnsureNoReconnect(String appName, String targetUrl) {
        for (Connection conn : getConnectionFactory(appName).allConnectionSnapshot()) {
            Client client = conn.getClient();
            if (targetUrl.equals(client.getConnectionUrl())) {
                client.destroy("ClientManager.closeEnsureNoReconnect()");
            }
        }
    }

    public void stopAllReconnectTasks() throws InterruptedException {
        this.reconnectMgr.stopAllTasks();
    }

    public int reconnectTasks() {
        return this.reconnectMgr.taskSize();
    }

    public Map<String,ClientConnectionFactory> clientFactorySnapshot() throws ExecutionException,
            InterruptedException {
        return this.repoMgr.clientFactorySnapshot();
    }

    @Override
    public void closeAllClients(String appName) {
        if (StringUtils.isBlank(appName)) {
            try {
                for (Map.Entry<String,ClientConnectionFactory> ccf :
                        this.repoMgr.clientFactorySnapshot().entrySet()) {
                    ccf.getValue().closeAllClients();
                }
            } catch (InterruptedException | ExecutionException e) {
                LoggerInit.LOGGER.warn("关闭连接工厂连接时出错",e.getCause());
            }
        } else {
            this.getConnectionFactory(appName).closeAllClients();
        }
    }

    @Override
    public ClientConnectionFactory getConnectionFactory(String appName) {
        return this.repoMgr.getConnectionFactory(appName);
    }

    private static List<IOEventListener> copyListener(List<IOEventListener> listeners,
                                                      Map<Class<?>,RequestProcessor<?>> processors) {
        List<IOEventListener> listenersCopy = new ArrayList<>();
        if (listeners != null) {
            listenersCopy.addAll(listeners);
        }

        if (processors != null) {
            IOEventListener.ConnectionMsgReceivedListener msgReceivedListener =
                    new DefaultMsgListener();
            for (RequestProcessor<?> p : processors.values()) {
                msgReceivedListener.registerProcessor(p);
            }

            listenersCopy.add(msgReceivedListener);
        }
        return listenersCopy;
    }

    static {
        timer.scheduleWithFixedDelay(new BackDoorDoggie(),0L,30L, TimeUnit.SECONDS);
    }
}
